package com.lzt.app

import java.util.Map.Entry

import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Result, Scan, Table}
import org.apache.hadoop.hbase.util.Bytes
import org.json4s.JsonAST.JValue
import org.slf4j.{Logger, LoggerFactory}

import scala.collection.JavaConverters.asScalaIteratorConverter
import scala.collection.mutable

object HbaseTemplate {

  private val logger: Logger = LoggerFactory.getLogger("HbaseTemplate")

  def getConn(): Connection = {
    val hbaseConfig = HBaseConfiguration.create()
    hbaseConfig.set("hbase.zookeeper.quorum", Config.zookeeperHost)
    val conn = ConnectionFactory.createConnection(hbaseConfig)
    logger.debug(s"create hbase connection : $conn")
    conn
  }

  /**
    * hbase 操作函数
    *
    * @param fun
    * @return
    */
  def operator[T](fun: Connection => T): T = {
    var conn: Connection = null
    try {
      conn = getConn()
      fun(conn)
    } catch {
      case e: Exception => {
        logger.error("hbase 操作异常：", e)
        throw e
      }
    } finally {
      conn.close()
    }
  }

  implicit def convertMap(x: Result): Map[String, String] = {
    val cellMap = x.getFamilyMap(Bytes.toBytes("cf")).entrySet().toArray().map(entry => {
      val row = entry.asInstanceOf[Entry[Array[Byte], Array[Byte]]]
      (new String(row.getKey), new String(row.getValue))
    }).toMap
    cellMap + ("rowkey" -> new String(x.getRow))
  }

  implicit def convertJson(x:Result): JValue = {
    import org.json4s._
    import org.json4s.JsonDSL._
    import org.json4s.jackson.JsonMethods._

    val cellJvalue = x.getFamilyMap(Bytes.toBytes("cf")).entrySet().toArray().map(entry => {
      val row = entry.asInstanceOf[Entry[Array[Byte], Array[Byte]]]
      (new String(row.getKey) -> JString(new String(row.getValue)))
    })
    implicit val formats = DefaultFormats
    render(cellJvalue.toList :+ ("rowkey" -> JString(new String(x.getRow))))
  }

  def scan[T](tableName: String, start: String, end: String)(implicit convert: Result => T):mutable.ArraySeq[T] = {
    println("scan table:{}, start:{}, end:{}" + List(tableName, start, end))
    operator [mutable.ArraySeq[T]]{ con =>
      val table: Table = con.getTable(TableName.valueOf(tableName))
      val scaner = new Scan()
      if (StringUtil.notNull(start)) {
        scaner.setStartRow(Bytes.toBytes(start))
      }
      if (StringUtil.notNull(end)) {
        scaner.setStopRow(Bytes.toBytes(end))
      }
      var results = table.getScanner(scaner)
      asScalaIteratorConverter(results.iterator())
        .asScala
        .toArray
        .map(convert)
    }
  }

  def scanWithMap(tableName: String, start: String, end: String) = {
    scan(tableName,start,end)(convertMap)
  }

//  def scan(tableName: String, start: String, end: String): Array[Map[String, String]] = {
//    operator { con =>
//      val table: Table = con.getTable(TableName.valueOf(tableName))
//      val scaner = new Scan()
//      if (StringUtil.notNull(start)) {
//        scaner.setStartRow(Bytes.toBytes(start))
//      }
//      if (StringUtil.notNull(end)) {
//        scaner.setStopRow(Bytes.toBytes(end))
//      }
//      var results = table.getScanner(scaner)
//      asScalaIteratorConverter(results.iterator())
//        .asScala
//        .toArray
//        .map(x => {
//          val cellMap = x.getFamilyMap(Bytes.toBytes("cf")).entrySet().toArray().map(entry => {
//            val row = entry.asInstanceOf[Entry[Array[Byte], Array[Byte]]]
//            (new String(row.getKey), new String(row.getValue))
//          }).toMap
//          cellMap + ("rowkey" -> new String(x.getRow))
//        })
//    }.asInstanceOf[Array[Map[String, String]]]
//  }
}
